package com.xquant.system.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xquant.system.common.ResultCode;
import com.xquant.system.entity.TwcBndPrimaryKeyTest;
import com.xquant.system.exception.ServiceException;
import com.xquant.system.mapper.TwcBndPrimaryKeyTestMapper;
import com.xquant.system.service.TwcBndPrimaryKeyTestService;
import com.xquant.system.thread.SelectTask;
import lombok.extern.log4j.Log4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.object.BatchSqlUpdate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;


/**
 * @author 04641
 * @description 针对表【TWC_BND_PRIMARY_KEY_TEST】的数据库操作Service实现  内码融合
 * 衡泰内码为准（各个数据厂商的原始内码不一致SOURCE_CODE） 数据来源 HT 1  ch 2 wd 3
 * if根据交易代码 + 交易市场
 * else根据交易市场 + 债券全称
 * @createDate 2023-09-05 11:03:49
 */
@Service
@Transactional(rollbackFor = Exception.class)
@Log4j
public class TwcBndPrimaryKeyTestServiceImpl extends ServiceImpl<TwcBndPrimaryKeyTestMapper, TwcBndPrimaryKeyTest>
        implements TwcBndPrimaryKeyTestService {


    @Autowired
    private TwcBndPrimaryKeyTestMapper twcBndPrimaryKeyTestMapper;
    @Autowired
    private DataSource dataSource;

    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 5, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
    /**
     * 内码融合
     * */
    @Override
    public String internalCodeFusion() {
        // 开始时间
        long startTime = System.currentTimeMillis();
        //条件组合
        Map<String, TwcBndPrimaryKeyTest> twcBndPrimaryKeyConditionOne = new HashMap<>();
        Map<String, TwcBndPrimaryKeyTest> twcBndPrimaryKeyConditionTwo = new HashMap<>();
        //处理后的结果集
        List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestList = new ArrayList<>();

        //三个线程分别查询三个数据源的数据
        List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestsHt;
        List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestsCh;
        List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestsWd;

        // 向线程池递交线程任务，1代表衡泰 2代表CH 3代表WD
        Future<List<TwcBndPrimaryKeyTest>> resultHt = threadPool.submit(new SelectTask(twcBndPrimaryKeyTestMapper,1));
        Future<List<TwcBndPrimaryKeyTest>> resultCh = threadPool.submit(new SelectTask(twcBndPrimaryKeyTestMapper,2));
        Future<List<TwcBndPrimaryKeyTest>> resultWd = threadPool.submit(new SelectTask(twcBndPrimaryKeyTestMapper,3));

        //获取线程执行结果
        try {
            twcBndPrimaryKeyTestsHt = resultHt.get();
            twcBndPrimaryKeyTestsCh = resultCh.get();
            twcBndPrimaryKeyTestsWd = resultWd.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }

        //以衡泰为基准 将条件封装
        twcBndPrimaryKeyTestsHt.forEach(twcBndPrimaryKeyTest -> {
            //如果衡泰的SEC_CODE为空 则将SOURCE_CODE赋值给SEC_CODE
            if(twcBndPrimaryKeyTest.getSEC_CODE()==null){
                twcBndPrimaryKeyTest.setSEC_CODE(twcBndPrimaryKeyTest.getSOURCE_CODE());
                //添加到结果集
                twcBndPrimaryKeyTestList.add(twcBndPrimaryKeyTest);
            }
            //根据交易代码 + 交易市场作为key 衡泰数据对象为vue 作为条件
            twcBndPrimaryKeyConditionOne.put(twcBndPrimaryKeyTest.getI_CODE() + twcBndPrimaryKeyTest.getM_TYPE(), twcBndPrimaryKeyTest);
            //根据交易市场 + 债券全称key 衡泰数据对象为vue  作为条件
            twcBndPrimaryKeyConditionTwo.put(twcBndPrimaryKeyTest.getM_TYPE() + twcBndPrimaryKeyTest.getNAME_FULL(), twcBndPrimaryKeyTest);
        });
        //衡泰标准填写CH和WD SEC_CODE和D_CODE 填充财汇
        dataMatching(twcBndPrimaryKeyTestsCh, twcBndPrimaryKeyTestList, twcBndPrimaryKeyConditionOne, twcBndPrimaryKeyConditionTwo);
        //衡泰标准填写CH和WD SEC_CODE和D_CODE 填充万得
        dataMatching(twcBndPrimaryKeyTestsWd, twcBndPrimaryKeyTestList, twcBndPrimaryKeyConditionOne, twcBndPrimaryKeyConditionTwo);
        //查询结束时间
        long queryEndTime = System.currentTimeMillis();

        //批量修改数据处理完成的数据
        updateData(twcBndPrimaryKeyTestList);

        //修改结束时间
        long updateEndTime = System.currentTimeMillis();
        String time = "查询执行时长：" + (queryEndTime - startTime) + "毫秒."
                + "更新执行时长：" + (updateEndTime - queryEndTime) + "毫秒."
                +"总执行时长：" + (updateEndTime - startTime) + "毫秒.";
        log.info("数据融合完成...................................");
        return time;
    }

    private void updateData(List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestList) {

        BatchSqlUpdate batchSqlUpdate =
                new BatchSqlUpdate(dataSource,
                        "UPDATE TWC_BND_PRIMARY_KEY_TEST SET SEC_CODE = ? , D_CODE = ? WHERE ID = ?");

        batchSqlUpdate.setBatchSize(200000);
        batchSqlUpdate.setTypes(new int[]{java.sql.Types.VARCHAR, java.sql.Types.VARCHAR, java.sql.Types.BIGINT});
        for (TwcBndPrimaryKeyTest bndPrimaryKeyTest : twcBndPrimaryKeyTestList) {
            batchSqlUpdate.update(new Object[]{bndPrimaryKeyTest.getSEC_CODE(), bndPrimaryKeyTest.getD_CODE(), bndPrimaryKeyTest.getID()});
        }
        try{
            batchSqlUpdate.flush();
        }catch (Exception e){
            throw new ServiceException(ResultCode.ERROR,"更新失败");
        }
    }

    //数据匹配 通过交易代码 + 交易市场 或 交易市场 + 债券全称
    public void dataMatching(List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTests,
                             List<TwcBndPrimaryKeyTest> twcBndPrimaryKeyTestList,
                             Map<String, TwcBndPrimaryKeyTest> twcBndPrimaryKeyConditionOne,
                             Map<String, TwcBndPrimaryKeyTest> twcBndPrimaryKeyConditionTwo) {

        for (TwcBndPrimaryKeyTest twcBndPrimaryKeyTest : twcBndPrimaryKeyTests) {
            if (twcBndPrimaryKeyConditionOne.get(twcBndPrimaryKeyTest.getI_CODE() + twcBndPrimaryKeyTest.getM_TYPE()) != null) {
                twcBndPrimaryKeyTest.setSEC_CODE(twcBndPrimaryKeyConditionOne.get(twcBndPrimaryKeyTest.getI_CODE() + twcBndPrimaryKeyTest.getM_TYPE()).getSOURCE_CODE());
                twcBndPrimaryKeyTest.setD_CODE(twcBndPrimaryKeyConditionOne.get(twcBndPrimaryKeyTest.getI_CODE() + twcBndPrimaryKeyTest.getM_TYPE()).getSOURCE_D_CODE());
                //添加到结果集
                twcBndPrimaryKeyTestList.add(twcBndPrimaryKeyTest);
            } else if (twcBndPrimaryKeyConditionTwo.get(twcBndPrimaryKeyTest.getM_TYPE()+ twcBndPrimaryKeyTest.getNAME_FULL()) != null) {
                twcBndPrimaryKeyTest.setSEC_CODE(twcBndPrimaryKeyConditionTwo.get(twcBndPrimaryKeyTest.getM_TYPE() + twcBndPrimaryKeyTest.getNAME_FULL()).getSOURCE_CODE());
                twcBndPrimaryKeyTest.setD_CODE(twcBndPrimaryKeyConditionTwo.get(twcBndPrimaryKeyTest.getM_TYPE() + twcBndPrimaryKeyTest.getNAME_FULL()).getSOURCE_D_CODE());
                //添加到结果集
                twcBndPrimaryKeyTestList.add(twcBndPrimaryKeyTest);
            }else {
                log.info("未匹配到数据");
                continue;
            }
        }
    }


}







